package org.kfive.nova.hadoop.reducer;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/*
* MultiTableJoinReducer.java
* Created  by  lizhanwu
* Created on 2016/4/20 19:49
*/
public class MultiTableJoinReducer extends Reducer<Text,Text,Text,Text> {

    public void reduce(Text key,Iterable<Text> values,Context context)throws IOException,InterruptedException{

        List<String> factoryList = new ArrayList<String>();
        List<String> addressList = new ArrayList<String>();

        for (Text value : values){
            String valueStr = value.toString();
            String type = valueStr.substring(0,1);
            if("1".equals(type)){
                factoryList.add(valueStr.substring(1));
            }else {
                addressList.add(valueStr.substring(1));
            }
        }

        for (String factory : factoryList){
            for (String address : addressList){
                context.write(new Text(factory),new Text(address));
            }
        }
    }

}
